package com.flink.datastreamapi.function;

import com.flink.datastreamapi.bean.WaterSensor;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

//富函数
public class RichFilterFunctionUDFDemo {

    public static void main(String[] args) throws Exception {

        //获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //获取数据源
        DataStreamSource<WaterSensor> stream = env.fromElements(

                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_1", 12L, 12),
                new WaterSensor("sensor_2", 2L, 2),
                new WaterSensor("sensor_3", 3L, 3)
        );

        //转换--过滤

        stream.filter(new RichFilterFunction<WaterSensor>() {

            //开始，可以在这里进行初始化
            @Override
            public void open(OpenContext openContext) throws Exception {

                System.out.println("=====open()==========");
            }

            @Override
            public boolean filter(WaterSensor value) throws Exception {
                System.out.println("=====filter()=========="+value);

                return "sensor_2".equals(value.id);
            }

            //结束---清理数据
            @Override
            public void close() throws Exception {

                System.out.println("=====close()==========");
            }
        })
        //打印
        .print();

       //执行
        env.execute();
    }


}
